package cn.doitedu.dwetl

import org.apache.spark.sql.SparkSession

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-01-17
 * @desc 流量基础指标，多维分析计算（利用bitmap做逐层聚合）
 *       目标表建表（bitmap数据结构的表）
 *   create table dws.tfc_overview_bitmap(
 *       province           string,
 *       city               string,
 *       region             string,
 *       hour_range         int   ,
 *       device_type        string,
 *       os_name            string,
 *       start_page         string,
 *       end_page           string,
 *       isnew              string,
 *       pv_cnt             bigint,
 *       uv_bitmap             binary,
 *       session_cnt        bigint,
 *       time_long          bigint,
 *       return_user_bitmap    binary,
 *       jumpout_cnt        bigint,
 *       jumpout_user_bitmap   binary
 *       )
 *    PARTITIONED BY (dt string)
 *    STORED AS PARQUET
 *    ;
 *
 *
 *
 */
object TrafficBasicMetricBitmapCalc {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("多维分析计算（利用bitmap做逐层聚合）")
      .enableHiveSupport()
      //.master("local[*]")
      .getOrCreate()
    import org.apache.spark.sql.functions._
    import spark.implicits._


    spark.sql(
      s"""
        |
        |select
        |cast(b.id as int) as id ,
        |a.province     ,
        |a.city         ,
        |a.region       ,
        |a.hour_range   ,
        |a.device_type  ,
        |a.os_name      ,
        |a.start_page   ,
        |a.end_page     ,
        |a.isnew        ,
        |a.pv_cnt       ,   -- 用户的当日pv总数
        |a.session_cnt  ,   -- 用户的会话总数
        |a.time_long    ,   -- 用户的总访问时长
        |a.jumpout_cnt      -- 用户的跳出次数
        |
        |from (select * from dws.tfc_app_agr_user where dt='${args(0)}') a  left join  dwd.user_guid_global b
        |on a.guid=b.guid
        |
        |""".stripMargin).createTempView("tmp")

    import cn.doitedu.dwetl.utils.RrUtils._
    spark.udf.register("to_bitmap",toBitmap)
    spark.sql(
      s"""
        |
        |INSERT INTO TABLE dws.tfc_overview_bitmap PARTITION(dt='${args(0)}')
        |SELECT
        |  province     ,   -- 省
        |  city         ,   -- 市
        |  region       ,
        |  hour_range   ,
        |  device_type  ,
        |  os_name      ,
        |  start_page   ,
        |  end_page     ,
        |  isnew        ,
        |  sum(pv_cnt) as pv_cnt,
        |
        |  to_bitmap(collect_set(id)) as uv_bitmap,
        |-- count(distinct guid) as uv_cnt,
        |
        |  sum(session_cnt) as session_cnt,
        |  sum(time_long) as time_long,
        |
        |  to_bitmap(collect_set(if(session_cnt>1,id,null))) as return_user_bitmap,
        |-- count(distinct if(session_cnt>1,guid,null)) as return_user_cnt,
        |
        |  sum(jumpout_cnt)  as jumpout_cnt,
        |  to_bitmap(collect_set( if(jumpout_cnt>0,id,null) ))  as jumpout_user_bitmap
        |--count(distinct if(jumpout_cnt>0,guid,null)) as jumpout_user_cnt
        |
        |FROM tmp
        |GROUP BY
        |  province     ,
        |  city         ,
        |  region       ,
        |  hour_range   ,
        |  device_type  ,
        |  os_name      ,
        |  start_page   ,
        |  end_page     ,
        |  isnew
        |
        |""".stripMargin)



    spark.close()

  }

}
